package yuan.consumer.rebalancer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import yuan.consumer.KafkaConsumerAnalysis;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @version: 1.0
 * @Description: TODO
 * @author: zZmH
 * @date: 2020/09/09 12:44
 */


public class ConsumerRebalance {

    static String topic = "topic-demo";
    public static final AtomicBoolean isRunning = new AtomicBoolean(true);



    public static void main(String[] args) {
        final Properties prop = KafkaConsumerAnalysis.init();
        final KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(prop);
        final HashMap<TopicPartition, OffsetAndMetadata> lastOffsetMap = new HashMap<>();

        consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() {

            /**
             * @author: zZmH
             * @description: 在消费者停止读取消息之后和再均衡之前触发
             * @date: 12:48 2020/9/9 0009
             * @param collection
             * @return void
             **/
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                consumer.commitSync(lastOffsetMap);
                lastOffsetMap.clear();
            }


            //在再均衡之后和消费者开始消费之前触发
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {


            }
        });

        while (isRunning.get()) {

            final ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> record : records) {
                lastOffsetMap.put(
                        new TopicPartition(record.topic(), record.partition()),
                        new OffsetAndMetadata(record.offset() + 1));

            }

            consumer.commitAsync(lastOffsetMap,null);

        }


    }


}
